package com.edu.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/19.
 *
 * @author Topus
 * @desc 考试域：试卷选项明细
 */
public class DwdTestPaperOption {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_test_paper_option_group"));

        Table resultTb = tableEnv.sqlQuery("select data['id'] id, data['question_id'] question_id,\n"
            + "data['is_correct'] is_correct, data['deleted'] deleted, ts\n" + "from topic_db\n"
            + "where `table`='test_question_option' and `type`='insert'");
        tableEnv.createTemporaryView("result_table", resultTb);

        tableEnv.executeSql("create table dwd_test_question_option(\n" + "id string,\n" + "question_id string,\n"
            + "is_correct string,\n" + "deleted string,\n" + "ts string,\n" + "primary key(id) not enforced\n" + ")"
            + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_question_option"));

        tableEnv.executeSql("insert into dwd_test_question_option select * from result_table");
    }
}
